{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "proprietary-inspiration",
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyflink.table import StreamTableEnvironment, EnvironmentSettings,TableEnvironment"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "respected-acoustic",
   "metadata": {},
   "outputs": [
    {
     "ename": "AttributeError",
     "evalue": "type object 'TableEnvironment' has no attribute 'create'",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mAttributeError\u001b[0m                            Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-15-b1b7d82aae0b>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      1\u001b[0m \u001b[0msettings\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mEnvironmentSettings\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mnew_instance\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0min_batch_mode\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0muse_blink_planner\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mbuild\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 2\u001b[0;31m \u001b[0mt_env\u001b[0m \u001b[0;34m=\u001b[0m \u001b[0mTableEnvironment\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mcreate\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0msettings\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m",
      "\u001b[0;31mAttributeError\u001b[0m: type object 'TableEnvironment' has no attribute 'create'"
     ]
    }
   ],
   "source": [
    "settings = EnvironmentSettings.new_instance().in_batch_mode().use_blink_planner().build()\n",
    "t_env = TableEnvironment.create(settings)"
   ]
  },
  {
   "cell_type": "raw",
   "id": "rural-supply",
   "metadata": {},
   "source": []
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "id": "chinese-junction",
   "metadata": {},
   "outputs": [
    {
     "ename": "ModuleNotFoundError",
     "evalue": "No module named 'pyflink.table.expressions'",
     "output_type": "error",
     "traceback": [
      "\u001b[0;31m---------------------------------------------------------------------------\u001b[0m",
      "\u001b[0;31mModuleNotFoundError\u001b[0m                       Traceback (most recent call last)",
      "\u001b[0;32m<ipython-input-9-dfb63d1f2f8c>\u001b[0m in \u001b[0;36m<module>\u001b[0;34m\u001b[0m\n\u001b[1;32m      1\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0mpyflink\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtable\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mDataTypes\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mEnvironmentSettings\u001b[0m\u001b[0;34m,\u001b[0m \u001b[0mStreamTableEnvironment\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0;32m----> 2\u001b[0;31m \u001b[0;32mfrom\u001b[0m \u001b[0mpyflink\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtable\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mexpressions\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mcol\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[0m\u001b[1;32m      3\u001b[0m \u001b[0;32mfrom\u001b[0m \u001b[0mpyflink\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mtable\u001b[0m\u001b[0;34m.\u001b[0m\u001b[0mudf\u001b[0m \u001b[0;32mimport\u001b[0m \u001b[0mudf\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      4\u001b[0m \u001b[0;34m\u001b[0m\u001b[0m\n\u001b[1;32m      5\u001b[0m \u001b[0;32mdef\u001b[0m \u001b[0mtable_api_demo\u001b[0m\u001b[0;34m(\u001b[0m\u001b[0;34m)\u001b[0m\u001b[0;34m:\u001b[0m\u001b[0;34m\u001b[0m\u001b[0m\n",
      "\u001b[0;31mModuleNotFoundError\u001b[0m: No module named 'pyflink.table.expressions'"
     ]
    }
   ],
   "source": [
    "from pyflink.table import DataTypes, EnvironmentSettings, StreamTableEnvironment \n",
    "from pyflink.table.expressions import col \n",
    "from pyflink.table.udf import udf \n",
    "\n",
    "def table_api_demo(): \n",
    "    env_settings = EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build() \n",
    "    t_env = StreamTableEnvironment.create(environment_settings=env_settings) \n",
    "    t_env.get_config().get_configuration().set_string('parallelism.default', '4') \n",
    "    t_env.execute_sql(\"\"\" CREATE TABLE my_source ( a VARCHAR, b VARCHAR ) WITH ( 'connector' = 'datagen', 'number-of-rows' = '10' ) \"\"\") \n",
    "    tab = t_env.from_path('my_source') @udf(result_type=DataTypes.STRING()) \n",
    "    \n",
    "def sub_string(s: str, begin: int, end: int): \n",
    "    return s[begin:end] \n",
    "\n",
    "transformed_tab = tab.select(sub_string(col('a'), 2, 4)) \n",
    "t_env.execute_sql(\"\"\" CREATE TABLE my_sink ( `sum` VARCHAR ) WITH ( 'connector' = 'print' ) \"\"\") \n",
    "table_result = transformed_tab.execute_insert('my_sink') \n",
    "# 1）等待作业执行结束，用于local执行，否则可能作业尚未执行结束，该脚本已退出，会导致minicluster过早退出 \n",
    "# 2）当作业通过detach模式往remote集群提交时，比如YARN/Standalone/K8s等，需要移除该方法 table_result.wait() \n",
    "if __name__ == '__main__': \n",
    "    table_api_demo()\n",
    "    "
   ]
  },
  {
   "cell_type": "raw",
   "id": "convertible-interaction",
   "metadata": {},
   "source": [
    "一、"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "geological-strain",
   "metadata": {},
   "outputs": [],
   "source": [
    "env_settings = EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build() \n",
    "t_env = StreamTableEnvironment.create(environment_settings=env_settings)\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.0"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}
